#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/resource.h>
#include <signal.h>
#include <pthread.h>

#include "error_code.h"
#include "fastcgi.h"
#include "parser.h"
#include "command.h"
#include "datainit.h"

#include "json/json.h"
#include "http_echo.h"
#include "hd_fs_operation.h"
#include "async_fs_operator.h"
#include "threadpool.h"

#define SERVER_THREAD_NUM 10
typedef void (*sig_handler)(int);
static int exit_flag = 0;
static pthread_mutex_t fcgx_mutex = PTHREAD_MUTEX_INITIALIZER;

HadoopFSOperation* g_fsops = NULL;
AsyncFSOperator<HadoopFSOperation, ThreadPool> *g_async_fsops = NULL;
ThreadPool *g_threadpool = NULL;

extern char **environ;
using namespace std;

void CloseRequest(FCGX_Request *request);
/*
int http_echo(char *content, int length)
{
    printf("Content-Type: text/html\r\n");
    printf("Content-Length: %d\r\n", length);
    printf("\r\n");

    printf("%s", content);
}*/

static void *
worker(void *arg)
{
	int ret;
    ParserStruc *pstruc = NULL;
    
    signal_mask(SIGALRM);
	signal_mask(SIGTERM);
	while (exit_flag == 0) {
    	FCGX_Request *request = new FCGX_Request();
        DINFO("Create request:%p", request);
	    FCGX_InitRequest(request, FCGI_LISTENSOCK_FILENO, FCGI_FAIL_ACCEPT_ON_INTR);
	
        pthread_mutex_lock(&fcgx_mutex);
		signal_unmask(SIGTERM);
		ret = FCGX_Accept_r(request);
		signal_mask(SIGTERM);
		pthread_mutex_unlock(&fcgx_mutex);

		if (exit_flag == 1)
			break;

		if (ret < 0) {
			if (exit_flag == 0) {
				DERROR("fcgi accept error, ignore, ret=%d\n", ret);
                
                CloseRequest(request);
                continue;
			} else {
				break;
			}
		}

        DINFO("script_name: %s\n", FCGX_GetParam("SCRIPT_NAME", request->envp));	
        DINFO("query_string: %s\n", FCGX_GetParam("QUERY_STRING", request->envp));	
        
		DINFO("start process command...\n");
		pstruc = parser_init();    
		ret = parser_requ_env(request, pstruc);
		if (ret < 0) {
			DERROR("parser_requ_env error\n");

			char errbuf[1024];
			dp_strerror(ret, errbuf, 1024);
			http_echo(request, 411, errbuf, strlen(errbuf));

			parser_destroy(pstruc);
            CloseRequest(request);
			continue;
		}
		ret = parser_storage_cmd(pstruc);
		if (ret < 0) {
			DERROR("parser_storage_cmd error\n");

			char errbuf[1024];
			dp_strerror(ERROR_CODE_ILLEGAL_ARGUMENT, errbuf, 1024);
			http_echo(request, 418, errbuf, strlen(errbuf));

			parser_destroy(pstruc);
		    CloseRequest(request);
			continue;
		}
        
		ret = process_cmd(request, pstruc);
        parser_destroy(pstruc);
	}
	DINFO("thread quit\n");
	//FCGX_Finish_r(&request);
	return NULL;
}

int
create_workers(pthread_t *t, int t_num)
{
	int i;
	pthread_attr_t attr;
	int ret;
	char errbuf[1024];

	ret = pthread_attr_init(&attr);
	if (ret != 0) {
		strerror_r(ret, errbuf, 1024);
		DERROR("pthread_attr_init error: %s\n", errbuf);
		return -1;
	}
	ret = pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
	if (ret != 0) {
		strerror_r(ret, errbuf, 1024);
		DERROR("pthread_attr_setscope error: %s\n", errbuf);
		pthread_attr_destroy(&attr);
		return -1;
	}
	ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
	if (ret != 0) {
		strerror_r(ret, errbuf, 1024);
		DERROR("pthread_attr_setdetachstate error: %s\n", errbuf);
		pthread_attr_destroy(&attr);
		return -1;
	}
	for (i = 0; i < t_num; i++) {
		int ret = pthread_create(&t[i], &attr, worker, NULL);
		if (ret != 0) {
			strerror_r(ret, errbuf, 1024);
			DERROR("pthread_create error: %s\n", errbuf);
			pthread_attr_destroy(&attr);
			return -1;
		}

	}
	pthread_attr_destroy(&attr);
	return 0;
}

int
dataserver()
{
    int ret;
/*
    ParserStruc *pstruc = NULL;
    
    DINFO("==wait accept\n");
    while (FCGI_Accept() >= 0) {
        pstruc = parser_init();    
        parser_requ_env(pstruc);
        ret = parser_storage_cmd(pstruc);
        ret = data_apply(pstruc);
        parser_destroy(pstruc);

    }
*/

	int i;
	FCGX_Init();
	pthread_t t_id[SERVER_THREAD_NUM-1];
	ret = create_workers(t_id, SERVER_THREAD_NUM-1);
	if (ret < 0) {
            return -1;
	}
	worker(NULL);
	DINFO("======wait threads quit======\n");
	for (i = 0; i < SERVER_THREAD_NUM-1; i++) {
		pthread_join(t_id[i], NULL);
	}
	DINFO("======dataserver done======\n");
    return 0;
}

void sighandler(int sig)
{
	DINFO("======get SIGTERM======\n");
	exit_flag = 1;
	close(FCGI_LISTENSOCK_FILENO);
}

void sig_segv(int sig)
{
	DERROR("======abort======\n");
	abort();
}

void setup_sighandler(int sig, sig_handler handler)
{
	struct sigaction siga;
	siga.sa_handler = handler;
	siga.sa_flags = 0;
	sigemptyset(&siga.sa_mask);
	sigaction(sig, &siga, NULL);
}

void setup_sighandlers()
{
	setup_sighandler(SIGTERM, sighandler);
	setup_sighandler(SIGSEGV, sig_segv);

	setup_sighandler(SIGPIPE, SIG_IGN);
}

int main(int argc, char **argv)
{
    int ret;
    struct rlimit rlim;

    g_log = logfile_create("test.log", 4);
    
    logfile_rotate_size(g_log, 40000000, 1);
    logfile_rotate_time(g_log, 14000000, 1);

	if (argc != 2) {
		DERROR("Usage: dataserver conf\n");
		exit(-1);
	}
    
    struct rlimit rlim_new;
    if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
        rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY;
        if (setrlimit(RLIMIT_CORE, &rlim_new) != 0) {
            rlim_new.rlim_cur = rlim_new.rlim_max = rlim.rlim_max;
            (void)setrlimit(RLIMIT_CORE, &rlim_new);
        }
    }
    if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) {
        return -1;
    }

	setup_sighandlers();

    ret = dataserver_init(argv[1]);
    if (ret < 0) {
        DERROR("dataserver_init: %d\n", ret);
        return -1;
    }
    dataserver();
    //cmd_index_add("xxx_users", "age", "cao");
    //cmd_index_remove("xxx_users", "age");
    //cmd_query("xxx_users", "age", "10", 0 , 10);
    return 0;
}
